package bx.job.actors;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import bx.job.JobServerBootstrap;

import java.util.concurrent.CompletableFuture;

public class JobTriggerActor extends AbstractBehavior<JobTriggerActor.JobTriggerActorMsg> {
    private static final int perFetchLimit = 10;


    public JobTriggerActor(ActorContext<JobTriggerActorMsg> context) {
        super(context);
    }

    public static Behavior<JobTriggerActor.JobTriggerActorMsg> create() {
        return Behaviors.setup(JobTriggerActor::new);
    }

    @Override
    public Receive<JobTriggerActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(JobTriggerActor.TriggerJobMsg.class, triggerJobMsg -> {
            CompletableFuture.supplyAsync(() -> JobServerBootstrap.getTriggerStore().getTriggerListByBeforeTime(System.currentTimeMillis(), perFetchLimit))
                    .thenAccept(bxTriggerInfoList -> bxTriggerInfoList.forEach(triggerInfo ->
                            triggerJobMsg.replyTo().tell(new JobProcessManagerActor.IssueJobFromTriggerMsg(triggerInfo))));
            return this;
        }).build();
    }

    public interface JobTriggerActorMsg {
    }

    public static record TriggerJobMsg(
            ActorRef<JobProcessManagerActor.JobProcessManagerActorMsg> replyTo) implements JobTriggerActorMsg {

    }
}
